package com.atguigu.flink.cep;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.List;
import java.util.Map;

/**
 * Created by Smexy on 2023/3/3
 *
 *  ①定义模式
 *  ②把模式作用到流上
 *  ③调用方法从流中获取匹配到的数据。
 *
 *  注意： 模式会被流中的每一条数据都应用一次。
 *          需要基于事件时间计算，不支持处理时间。
 *
 *          只有水印超过当前数据的eventtime，数据才会触发运算。
 *          迟到的数据无法处理。
 */
public class Demo1_HelloWorld
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
            .<WaterSensor>forMonotonousTimestamps()
            .withTimestampAssigner( (e, ts) -> e.getTs());

        env.setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> ds = env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction())
            .assignTimestampsAndWatermarks(watermarkStrategy);

        //定义模式  begin("规则的名字")
        Pattern<WaterSensor, WaterSensor> pattern = Pattern.<WaterSensor>begin("s1")
            //规则的条件
            .where(new SimpleCondition<WaterSensor>()
            {
                //对数据进行匹配,返回true是匹配到的数据
                @Override
                public boolean filter(WaterSensor value) throws Exception {
                    return "s1".equals(value.getId());
                }
            });

        //把模式作用到流上,把普通流变为模式流
        PatternStream<WaterSensor> patternStream = CEP.pattern(ds, pattern);

        //获取模式流中匹配到的数据
        patternStream.select(new PatternSelectFunction<WaterSensor, String>()
        {
            /*
                每个匹配到的数据，会调用一次
                    Map<String, List<WaterSensor>> pattern
                        key： 匹配的规则名  s1
                        value:  匹配到的数据集合
             */
            @Override
            public String select(Map<String, List<WaterSensor>> pattern) throws Exception {
                return pattern.toString();
            }
        })
                     .print();


        try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

    }
}
